package com.chiaic.unittest.kafka;

import com.chiaic.unittest.common.thread.UnitTestThreadFactory;
import com.chiaic.unittest.data.ProtoType;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.Objects;
import java.util.concurrent.*;

import static java.util.concurrent.TimeUnit.SECONDS;

@Service
@NoArgsConstructor
@AllArgsConstructor
@Slf4j
public class KafkaProduceTestService {

    private ExecutorService dispatcherInExecutor;

    private ExecutorService dispatcherOutExecutor;

    private int dispatcherOutThreads;
    private int dispatcherInThreads;

    private final String IN = "queueIn";
    private final String OUT = "queueOut";

    private final String SIZE = "queueSize";

    private final String TO = "queueTimeout";

    private long batchSize = 100000;

    private long timeOut = 1000;

    private int sleepTime = 5;

    private ProtoType protoType = ProtoType.PROTO_BUFF;

    //    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    public KafkaProduceTestService(@Value("${kafka.dispatcherOutThreads:1}") int dispatcherOutThreads,
                                   @Value("${kafka.dispatcherInThreads:1}") int dispatcherInThreads,
                                   @Value("${kafka.batchSize:1}") long batchSize,
                                   @Value("${kafka.timeOut:50}") long timeOut,
                                   @Value("${kafka.sleepTime:50}") int sleepTime,
                                   @Value("${kafka.protoType:PROTO_BUFF}") ProtoType protoType) {
        log.info("KafkaProduceTestService Start");

        this.batchSize = batchSize;
        this.timeOut = timeOut;
        this.sleepTime = sleepTime;
        this.protoType = protoType;
        this.dispatcherInThreads = dispatcherInThreads;
        this.dispatcherOutThreads = dispatcherOutThreads;


        if (dispatcherInThreads > 0) {
            this.dispatcherInExecutor = new ThreadPoolExecutor(dispatcherInThreads,
                    dispatcherInThreads, 60, SECONDS,
                    new LinkedBlockingQueue<>(), UnitTestThreadFactory.forName("queueTest-In-"));
            for (int i = 0; i < dispatcherInThreads; i++) {
                dispatcherInExecutor.execute(this::dispatchIn);
            }
        } else {
            this.dispatcherInExecutor = null;
        }

        if (dispatcherOutThreads > 0) {
            this.dispatcherOutExecutor = new ThreadPoolExecutor(dispatcherOutThreads,
                    dispatcherOutThreads, 60, SECONDS,
                    new LinkedBlockingQueue<>(), UnitTestThreadFactory.forName("queueTest-Out-"));


            for (int i = 0; i < dispatcherOutThreads; i++) {

                dispatcherOutExecutor.submit(this::dispatchOut);
            }
        } else {
            this.dispatcherOutExecutor = null;
        }
    }

    private void dispatchIn() {
        log.info("start send to kafka");
        CompletableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("messagePackage", "test1");
        listenableFuture
                .whenComplete((stringStringSendResult, throwable) -> {
                    if (Objects.isNull(throwable)) {
                        log.info("kafka Send error");
                        throwable.printStackTrace();
                    } else {
                        log.info("send to kafka success");
                    }
                })
        // 下面设置出现异常时设置默认值，如果不需要出现异常时设置默认值则需要去掉(去掉时listenableFuture.get()会捕获到异常)
//                .exceptionally(throwable -> {
//                    log.info("kafka Send error");
//                    throwable.printStackTrace();
//                    return null;
//                })
        ;
//        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send("messagePackage", "test1");
//        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
//            @Override
//            public void onFailure(Throwable ex) {
//                log.info("kafka Send error");
//                ex.printStackTrace();
//            }
//
//            @Override
//            public void onSuccess(SendResult<String, String> result) {
//                log.info("send to kafka success");
//            }
//        });
        try {
            listenableFuture.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
//        Futures.addCallback(listenableFuture,new FutureCallback<SendResult<String,String>>(){
//
//            @Override
//            public void onSuccess(@NullableDecl SendResult<String, String> stringStringSendResult) {
//
//            }
//
//            @Override
//            public void onFailure(Throwable throwable) {
//
//            }
//        });
//    }

    private void dispatchOut() {

    }
}
